package com.xiongjie.ClusterLimit;

import com.xiongjie.ClusterLimit.Interface.IClusterLimit;
import com.xiongjie.ClusterLimit.entity.ClusterLimitEntity;
import com.xiongjie.ClusterLimit.listener.ClusterLimitManagerListener;
import com.xiongjie.ClusterLimit.service.SClusterLimitService;
import org.infinispan.Cache;
import org.infinispan.manager.DefaultCacheManager;
import org.infinispan.manager.EmbeddedCacheManager;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.IntStream;

public class ClusterLimitApplication implements Runnable {

    static final String[] names = {"Rome","Como","Basel"};

    private final EmbeddedCacheManager cacheManager;
    private final IClusterLimit iClusterCache;
    private Cache<String, Object> cache;
    private final ClusterLimitManagerListener cacheManagerListener;

    private ReentrantLock lock = new ReentrantLock(true);
    private ArrayBlockingQueue<Object> tokenBucket = new ArrayBlockingQueue<>(20);
    private ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    /**
     * 构造函数，创建缓存管理器和接口实现类
     */
    public ClusterLimitApplication() throws InterruptedException, IOException {

        cacheManager = new DefaultCacheManager(ClusterLimitApplication.class.getResourceAsStream("/cluster-infinispan.xml"));

        cacheManagerListener=new ClusterLimitManagerListener();
        cacheManager.addListener(cacheManagerListener);

        cache = cacheManager.getCache("clusterCache");
        cache.put("lock",lock);
        cache.put("tokenBucket",tokenBucket);

        iClusterCache = initClusterCacheService(cache);

    }

    /**
     * 销毁缓存管理器
     */
    public void shutdown() throws InterruptedException{
        cacheManager.stop();
    }

    /**
     * 初始化接口实现类，重要的数据，在系统环境变量中设置
     *
     * @param cache
     * @return
     */
    private IClusterLimit initClusterCacheService(Cache<String, Object> cache) {
        return new SClusterLimitService("", cache);
    }

    /**
     * 父类引用调用子类对象
     */
    public void getClusterCacheEntity() {
        System.out.println("获取分布式限流实体数据");

        long start = System.currentTimeMillis();
        for (String name : names) {
            ClusterLimitEntity entity = iClusterCache.getValueByName(name);
            if(entity!=null) {
                System.out.println(String.format("%s - %s", name, entity));
            }else{
                System.out.println(String.format("%s请求被限流了", name));
            }
        }

        System.out.println(String.format("获取分布式限流实体数据花费时间：%dms", System.currentTimeMillis() - start));
    }

    @Override
    public void run() {
        for (int i = 0; i < 5; i++) {
            tokenBucket.offer(new Object());
        }
        System.out.println("令牌数=" + tokenBucket.size());
        cache.put("tokenBucket",tokenBucket);
    }

    public static void main(String[] args) throws Exception {
        ClusterLimitApplication app = new ClusterLimitApplication();
        app.scheduledExecutorService.scheduleAtFixedRate(app, 0, 1, TimeUnit.SECONDS);


        //模拟0.1秒中发送一次请求,每次请求获取多个令牌
        IntStream.range(0, 2).forEach(i -> {
            new Thread(() -> {
                while (true) {
                    app.getClusterCacheEntity();
                    Random random=new Random();
                    int r=random.nextInt(2000);
                    try {
                        Thread.sleep(r);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        });
    }


}
